package com.example.aggregation;

import com.example.model.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created with IntelliJ IDEA.
 * ClassName: KeyByStream
 * Package: com.example.aggregation
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-22
 * Time: 16:31
 */


//聚合算子
public class KeyByStream {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //env.setParallelism(1);


        final SingleOutputStreamOperator<WaterSensor> map = env.socketTextStream("hadoop102", 7777)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        //切割
                        final String[] s = value.split(" ");

                        return new WaterSensor(s[0], Long.valueOf(s[1]), Integer.valueOf(s[2]));
                    }
                });


        //keyBy操作 按键分区 参数 key的选择器 的参数 输入的数据类型 选择key的数据类型
        //Tuple类型可以指定位置 或者多个位置的组合 POJO可以指定字段
        final KeyedStream<WaterSensor, String> keyData = map.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor value) throws Exception {
                return value.getId();
            }
        });
        //因为是对海量数据的处理 在聚合之前一定要对数据进行KeyBy分区操作 提交效率
        //KeyBy是从逻辑上划分为不用的分区 就是并行处理的子任务

        //聚合水位 相同的key的VC水位值进行相加
        //keyData.sum("vc").print();

        //max min 简单聚合算子 只能简单的聚合一个字段的值
        //minBy maxBy 通过判断指定的字段 返回符合要求的该字段值完整完整数据
        //keyData.minBy("vc").print();

        //reduce 规约聚合
        //一定是keyBy之后在调用 并且输入数据等于 输出数据
        //规约聚合的结果 作为一个状态保存起来 等下下一个数据来之后规约计算
        SingleOutputStreamOperator<WaterSensor> reduce = keyData.reduce(new ReduceFunction<WaterSensor>() {
            @Override
            public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                //相同的key才会进来 并且只来了一条数据 不会执行reduce 聚合至少两条数据 计算之后是一个结果值保存起来
                return new WaterSensor(value1.getId(), Long.valueOf(value1.getTs() + value2.getTs()), Integer.valueOf(value1.getVc()));
            }
        });

        reduce.print();

        env.execute();
    }
}
